//! 查询执行算子集成测试
//!
//! 测试火山模型算子的实际执行能力

use super::*;
use crate::storage::{StorageEngine, Table, TableSchema, ColumnDefinition, DataType};
use crate::sql::Value;
use std::sync::Arc;
use crate::config::DatabaseConfig;

/// 创建测试用的执行上下文
fn create_test_context() -> ExecutionContext {
    let config = DatabaseConfig::default();
    let storage_engine = StorageEngine::new(&config).unwrap();
    let page_manager = Arc::new(crate::storage::page::PageManager::new(100));

    let mut table_metadata = HashMap::new();
    table_metadata.insert("test_table".to_string(), TableMetadata {
        name: "test_table".to_string(),
        columns: vec![
            ColumnInfo::new("id".to_string(), DataType::Integer, false),
            ColumnInfo::new("name".to_string(), DataType::Text(Some(50)), false),
            ColumnInfo::new("age".to_string(), DataType::Integer, false),
        ],
        root_page_id: Some(1),
        row_count: 3,
        statistics: TableStatistics::default(),
    });

    ExecutionContext {
        page_manager,
        transaction_id: 1,
        table_metadata,
        statistics: ExecutionStatistics::default(),
        memory_limit: 1024 * 1024, // 1MB
        memory_usage: 0,
    }
}

/// 创建测试数据
fn create_test_data() -> Vec<Vec<Value>> {
    vec![
        vec![
            Value::Integer(1),
            Value::Text("Alice".to_string()),
            Value::Integer(25),
        ],
        vec![
            Value::Integer(2),
            Value::Text("Bob".to_string()),
            Value::Integer(30),
        ],
        vec![
            Value::Integer(3),
            Value::Text("Charlie".to_string()),
            Value::Integer(35),
        ],
    ]
}

/// 创建测试用的表数据
fn create_test_table() -> Table {
    let page_manager = Arc::new(crate::storage::page::PageManager::new(100));
    let schema = TableSchema::new(vec![
        ColumnDefinition {
            name: "id".to_string(),
            data_type: DataType::Integer,
            nullable: false,
            primary_key: true,
        },
        ColumnDefinition {
            name: "name".to_string(),
            data_type: DataType::Text(Some(50)),
            nullable: false,
            primary_key: false,
        },
        ColumnDefinition {
            name: "age".to_string(),
            data_type: DataType::Integer,
            nullable: false,
            primary_key: false,
        },
    ]);
    
    let mut table = Table::new("test_table".to_string(), schema, page_manager);
    
    // 插入测试数据
    table.insert_row(vec![
        Value::Integer(1),
        Value::Text("Alice".to_string()),
        Value::Integer(25),
    ]).unwrap();
    
    table.insert_row(vec![
        Value::Integer(2),
        Value::Text("Bob".to_string()),
        Value::Integer(30),
    ]).unwrap();
    
    table.insert_row(vec![
        Value::Integer(3),
        Value::Text("Charlie".to_string()),
        Value::Integer(35),
    ]).unwrap();
    
    table
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn test_table_scan_operator() {
        let mut context = create_test_context();
        let table = create_test_table();
        
        // 创建表扫描算子
        let schema = ExecutionSchema::new(vec![
            ColumnInfo::new("id".to_string(), DataType::Integer, false),
            ColumnInfo::new("name".to_string(), DataType::Text(Some(50)), false),
            ColumnInfo::new("age".to_string(), DataType::Integer, false),
        ]);
        
        let mut table_scan = TableScanOperator::new("test_table".to_string(), None, &context).unwrap();
        
        // 打开算子
        table_scan.open(&mut context).unwrap();
        
        // 读取所有元组
        let mut tuples = Vec::new();
        while let Some(tuple) = table_scan.next(&mut context).unwrap() {
            tuples.push(tuple);
        }
        
        // 验证结果
        assert_eq!(tuples.len(), 3);
        
        // 验证第一个元组
        assert_eq!(tuples[0].values[0], Value::Integer(1));
        assert_eq!(tuples[0].values[1], Value::Text("Alice".to_string()));
        assert_eq!(tuples[0].values[2], Value::Integer(25));
        
        // 关闭算子
        table_scan.close(&mut context).unwrap();
        
        // 验证统计信息
        let stats = table_scan.statistics();
        assert_eq!(stats.tuples_output, 3);
    }

    #[test]
    fn test_projection_operator() {
        let mut context = create_test_context();
        
        // 创建输入算子（表扫描）
        let schema = ExecutionSchema::new(vec![
            ColumnInfo::new("id".to_string(), DataType::Integer, false),
            ColumnInfo::new("name".to_string(), DataType::Text(Some(50)), false),
            ColumnInfo::new("age".to_string(), DataType::Integer, false),
        ]);
        
        let table_scan = Box::new(TableScanOperator::new("test_table".to_string(), None, &context).unwrap());
        
        // 创建投影算子（只选择id和name列）
        let projection_indices = vec![0, 1]; // id, name
        let mut projection = ProjectionOperator::new(table_scan, projection_indices).unwrap();
        
        // 打开算子
        projection.open(&mut context).unwrap();
        
        // 读取所有元组
        let mut tuples = Vec::new();
        while let Some(tuple) = projection.next(&mut context).unwrap() {
            tuples.push(tuple);
        }
        
        // 验证结果
        assert_eq!(tuples.len(), 3);
        assert_eq!(tuples[0].values.len(), 2); // 只有两列
        assert_eq!(tuples[0].values[0], Value::Integer(1));
        assert_eq!(tuples[0].values[1], Value::Text("Alice".to_string()));
        
        // 关闭算子
        projection.close(&mut context).unwrap();
        
        // 验证输出模式
        let output_schema = projection.schema();
        assert_eq!(output_schema.columns.len(), 2);
        assert_eq!(output_schema.columns[0].name, "id");
        assert_eq!(output_schema.columns[1].name, "name");
    }

    #[test]
    fn test_filter_operator() {
        let mut context = create_test_context();
        
        // 创建输入算子（表扫描）
        let schema = ExecutionSchema::new(vec![
            ColumnInfo::new("id".to_string(), DataType::Integer, false),
            ColumnInfo::new("name".to_string(), DataType::Text(Some(50)), false),
            ColumnInfo::new("age".to_string(), DataType::Integer, false),
        ]);
        
        let table_scan = Box::new(TableScanOperator::new("test_table".to_string(), None, &context).unwrap());
        
        // 创建过滤算子（age > 28）
        let predicate = FilterPredicate::Comparison {
            left_column: 2, // age列
            operator: ComparisonOperator::GreaterThan,
            right_value: Value::Integer(28),
        };
        
        let mut filter = FilterOperator::new(table_scan, predicate);
        
        // 打开算子
        filter.open(&mut context).unwrap();
        
        // 读取所有元组
        let mut tuples = Vec::new();
        while let Some(tuple) = filter.next(&mut context).unwrap() {
            tuples.push(tuple);
        }
        
        // 验证结果（应该只有Bob和Charlie，年龄>28）
        assert_eq!(tuples.len(), 2);
        assert_eq!(tuples[0].values[1], Value::Text("Bob".to_string()));
        assert_eq!(tuples[1].values[1], Value::Text("Charlie".to_string()));
        
        // 关闭算子
        filter.close(&mut context).unwrap();
        
        // 验证统计信息
        let stats = filter.statistics();
        assert_eq!(stats.tuples_processed, 3); // 处理了3个元组
        assert_eq!(stats.tuples_output, 2);    // 输出了2个元组
    }

    #[test]
    fn test_combined_operators() {
        let mut context = create_test_context();
        
        // 创建复合查询：SELECT id, name FROM test_table WHERE age > 25
        
        // 1. 表扫描
        let schema = ExecutionSchema::new(vec![
            ColumnInfo::new("id".to_string(), DataType::Integer, false),
            ColumnInfo::new("name".to_string(), DataType::Text(Some(50)), false),
            ColumnInfo::new("age".to_string(), DataType::Integer, false),
        ]);
        
        let table_scan = Box::new(TableScanOperator::new("test_table".to_string(), None, &context).unwrap());
        
        // 2. 过滤（age > 25）
        let predicate = FilterPredicate::Comparison {
            left_column: 2, // age列
            operator: ComparisonOperator::GreaterThan,
            right_value: Value::Integer(25),
        };
        
        let filter = Box::new(FilterOperator::new(table_scan, predicate));
        
        // 3. 投影（id, name）
        let projection_indices = vec![0, 1]; // id, name
        let mut projection = ProjectionOperator::new(filter, projection_indices).unwrap();
        
        // 执行查询
        projection.open(&mut context).unwrap();
        
        let mut results = Vec::new();
        while let Some(tuple) = projection.next(&mut context).unwrap() {
            results.push(tuple);
        }
        
        projection.close(&mut context).unwrap();
        
        // 验证结果（Bob和Charlie，只有id和name列）
        assert_eq!(results.len(), 2);
        assert_eq!(results[0].values.len(), 2);
        assert_eq!(results[0].values[0], Value::Integer(2));
        assert_eq!(results[0].values[1], Value::Text("Bob".to_string()));
        assert_eq!(results[1].values[0], Value::Integer(3));
        assert_eq!(results[1].values[1], Value::Text("Charlie".to_string()));
    }

    #[test]
    fn test_operator_reset() {
        let mut context = create_test_context();

        let schema = ExecutionSchema::new(vec![
            ColumnInfo::new("id".to_string(), DataType::Integer, false),
            ColumnInfo::new("name".to_string(), DataType::Text(Some(50)), false),
            ColumnInfo::new("age".to_string(), DataType::Integer, false),
        ]);

        let mut table_scan = TableScanOperator::new("test_table".to_string(), None, &context).unwrap();

        // 第一次执行
        table_scan.open(&mut context).unwrap();
        let mut count1 = 0;
        while table_scan.next(&mut context).unwrap().is_some() {
            count1 += 1;
        }
        table_scan.close(&mut context).unwrap();

        // 重置并再次执行
        table_scan.reset().unwrap();
        table_scan.open(&mut context).unwrap();
        let mut count2 = 0;
        while table_scan.next(&mut context).unwrap().is_some() {
            count2 += 1;
        }
        table_scan.close(&mut context).unwrap();

        // 两次执行应该得到相同的结果
        assert_eq!(count1, count2);
        assert_eq!(count1, 3);
    }

    #[test]
    fn test_hash_join_operator() {
        let mut context = create_test_context();

        // 创建左侧表扫描（用户表）
        let left_schema = ExecutionSchema::new(vec![
            ColumnInfo::new("user_id".to_string(), DataType::Integer, false),
            ColumnInfo::new("name".to_string(), DataType::Text(Some(50)), false),
        ]);

        let mut left_scan = TableScanOperator::new("users".to_string(), None, &context).unwrap();

        // 创建右侧表扫描（订单表）
        let right_schema = ExecutionSchema::new(vec![
            ColumnInfo::new("order_id".to_string(), DataType::Integer, false),
            ColumnInfo::new("user_id".to_string(), DataType::Integer, false),
            ColumnInfo::new("amount".to_string(), DataType::Real, false),
        ]);

        let mut right_scan = TableScanOperator::new("orders".to_string(), None, &context).unwrap();

        // 设置测试数据
        // 左侧算子返回用户数据
        left_scan.set_test_data(vec![
            ExecutionTuple::new(vec![Value::Integer(1), Value::Text("Alice".to_string())]),
            ExecutionTuple::new(vec![Value::Integer(2), Value::Text("Bob".to_string())]),
            ExecutionTuple::new(vec![Value::Integer(3), Value::Text("Charlie".to_string())]),
        ]);

        // 右侧算子返回订单数据
        right_scan.set_test_data(vec![
            ExecutionTuple::new(vec![Value::Integer(101), Value::Integer(1), Value::Real(100.0)]),
            ExecutionTuple::new(vec![Value::Integer(102), Value::Integer(2), Value::Real(200.0)]),
            ExecutionTuple::new(vec![Value::Integer(103), Value::Integer(1), Value::Real(150.0)]),
        ]);

        // 创建连接条件：users.user_id = orders.user_id
        let join_condition = JoinCondition {
            left_column: 0,  // users.user_id
            right_column: 1, // orders.user_id
            operator: ComparisonOperator::Equal,
        };

        // 创建哈希连接算子
        let mut hash_join = HashJoinOperator::new(
            Box::new(left_scan),
            Box::new(right_scan),
            join_condition,
            JoinType::Inner,
        );

        // 执行连接
        hash_join.open(&mut context).unwrap();

        let mut results = Vec::new();
        while let Some(tuple) = hash_join.next(&mut context).unwrap() {
            results.push(tuple);
        }

        hash_join.close(&mut context).unwrap();

        // 验证结果
        // 应该有3个连接结果：
        // (1, "Alice", 101, 1, 100.0)
        // (1, "Alice", 103, 1, 150.0)
        // (2, "Bob", 102, 2, 200.0)
        assert_eq!(results.len(), 3);

        // 验证第一个结果
        assert_eq!(results[0].values.len(), 5); // 2 + 3 列
        assert_eq!(results[0].values[0], Value::Integer(1));     // user_id
        assert_eq!(results[0].values[1], Value::Text("Alice".to_string())); // name
        assert_eq!(results[0].values[2], Value::Integer(101));   // order_id
        assert_eq!(results[0].values[3], Value::Integer(1));     // user_id
        assert_eq!(results[0].values[4], Value::Real(100.0));    // amount

        // 验证统计信息
        let stats = hash_join.statistics();
        assert_eq!(stats.tuples_output, 3);
    }

    #[test]
    fn test_aggregation_operator() {
        let mut context = create_test_context();

        // 创建输入表扫描（销售数据）
        let input_schema = ExecutionSchema::new(vec![
            ColumnInfo::new("product_id".to_string(), DataType::Integer, false),
            ColumnInfo::new("category".to_string(), DataType::Text(Some(50)), false),
            ColumnInfo::new("amount".to_string(), DataType::Real, false),
            ColumnInfo::new("quantity".to_string(), DataType::Integer, false),
        ]);

        let mut input_scan = TableScanOperator::new("sales".to_string(), None, &context).unwrap();

        // 设置测试数据：销售记录
        input_scan.set_test_data(vec![
            ExecutionTuple::new(vec![Value::Integer(1), Value::Text("Electronics".to_string()), Value::Real(100.0), Value::Integer(2)]),
            ExecutionTuple::new(vec![Value::Integer(2), Value::Text("Books".to_string()), Value::Real(50.0), Value::Integer(1)]),
            ExecutionTuple::new(vec![Value::Integer(3), Value::Text("Electronics".to_string()), Value::Real(200.0), Value::Integer(1)]),
            ExecutionTuple::new(vec![Value::Integer(4), Value::Text("Books".to_string()), Value::Real(30.0), Value::Integer(3)]),
            ExecutionTuple::new(vec![Value::Integer(5), Value::Text("Electronics".to_string()), Value::Real(150.0), Value::Integer(2)]),
        ]);

        // 创建聚合表达式：按category分组，计算COUNT、SUM(amount)、AVG(amount)
        let aggregate_expressions = vec![
            AggregateExpression {
                function: AggregateFunction::Count,
                column_index: None, // COUNT(*)
                output_name: "count".to_string(),
            },
            AggregateExpression {
                function: AggregateFunction::Sum,
                column_index: Some(2), // amount列
                output_name: "total_amount".to_string(),
            },
            AggregateExpression {
                function: AggregateFunction::Avg,
                column_index: Some(2), // amount列
                output_name: "avg_amount".to_string(),
            },
        ];

        // 创建输出模式
        let output_schema = ExecutionSchema::new(vec![
            ColumnInfo::new("category".to_string(), DataType::Text(Some(50)), false),
            ColumnInfo::new("count".to_string(), DataType::Integer, false),
            ColumnInfo::new("total_amount".to_string(), DataType::Real, false),
            ColumnInfo::new("avg_amount".to_string(), DataType::Real, false),
        ]);

        // 创建聚合算子：按category分组
        let mut aggregation = AggregationOperator::new(
            Box::new(input_scan),
            vec![1], // 按category列（索引1）分组
            aggregate_expressions,
            output_schema,
        );

        // 执行聚合
        aggregation.open(&mut context).unwrap();

        let mut results = Vec::new();
        while let Some(tuple) = aggregation.next(&mut context).unwrap() {
            results.push(tuple);
        }

        aggregation.close(&mut context).unwrap();

        // 验证结果
        // 应该有2个分组：Electronics和Books
        assert_eq!(results.len(), 2);

        // 验证每个结果元组的结构
        for result in &results {
            assert_eq!(result.values.len(), 4); // category + 3个聚合结果

            // 验证聚合结果的类型
            match &result.values[1] {
                Value::Integer(_) => {}, // count
                _ => panic!("COUNT结果应该是Integer类型"),
            }
            match &result.values[2] {
                Value::Real(_) => {}, // sum
                _ => panic!("SUM结果应该是Real类型"),
            }
            match &result.values[3] {
                Value::Real(_) => {}, // avg
                _ => panic!("AVG结果应该是Real类型"),
            }
        }

        // 验证统计信息
        let stats = aggregation.statistics();
        assert_eq!(stats.tuples_output, 2);
    }

    #[test]
    fn test_aggregation_without_groupby() {
        let mut context = create_test_context();

        // 创建输入表扫描
        let input_schema = ExecutionSchema::new(vec![
            ColumnInfo::new("value".to_string(), DataType::Integer, false),
        ]);

        let mut input_scan = TableScanOperator::new("numbers".to_string(), None, &context).unwrap();

        // 设置测试数据
        input_scan.set_test_data(vec![
            ExecutionTuple::new(vec![Value::Integer(10)]),
            ExecutionTuple::new(vec![Value::Integer(20)]),
            ExecutionTuple::new(vec![Value::Integer(30)]),
            ExecutionTuple::new(vec![Value::Integer(40)]),
            ExecutionTuple::new(vec![Value::Integer(50)]),
        ]);

        // 创建聚合表达式：不分组的全局聚合
        let aggregate_expressions = vec![
            AggregateExpression {
                function: AggregateFunction::Count,
                column_index: None,
                output_name: "total_count".to_string(),
            },
            AggregateExpression {
                function: AggregateFunction::Sum,
                column_index: Some(0),
                output_name: "total_sum".to_string(),
            },
            AggregateExpression {
                function: AggregateFunction::Min,
                column_index: Some(0),
                output_name: "min_value".to_string(),
            },
            AggregateExpression {
                function: AggregateFunction::Max,
                column_index: Some(0),
                output_name: "max_value".to_string(),
            },
        ];

        // 创建输出模式
        let output_schema = ExecutionSchema::new(vec![
            ColumnInfo::new("total_count".to_string(), DataType::Integer, false),
            ColumnInfo::new("total_sum".to_string(), DataType::Real, false),
            ColumnInfo::new("min_value".to_string(), DataType::Integer, false),
            ColumnInfo::new("max_value".to_string(), DataType::Integer, false),
        ]);

        // 创建聚合算子：无分组
        let mut aggregation = AggregationOperator::new(
            Box::new(input_scan),
            vec![], // 无分组列
            aggregate_expressions,
            output_schema,
        );

        // 执行聚合
        aggregation.open(&mut context).unwrap();

        let mut results = Vec::new();
        while let Some(tuple) = aggregation.next(&mut context).unwrap() {
            results.push(tuple);
        }

        aggregation.close(&mut context).unwrap();

        // 验证结果
        // 无分组聚合应该只有1个结果
        assert_eq!(results.len(), 1);

        let result = &results[0];
        assert_eq!(result.values.len(), 4);

        // 验证聚合结果
        assert_eq!(result.values[0], Value::Integer(5));    // COUNT = 5
        assert_eq!(result.values[1], Value::Real(150.0));   // SUM = 10+20+30+40+50 = 150
        assert_eq!(result.values[2], Value::Integer(10));   // MIN = 10
        assert_eq!(result.values[3], Value::Integer(50));   // MAX = 50

        // 验证统计信息
        let stats = aggregation.statistics();
        assert_eq!(stats.tuples_output, 1);
    }

    #[test]
    fn test_sort_operator() {
        let mut context = create_test_context();

        // 创建输入表扫描
        let input_schema = ExecutionSchema::new(vec![
            ColumnInfo::new("id".to_string(), DataType::Integer, false),
            ColumnInfo::new("name".to_string(), DataType::Text(Some(50)), false),
            ColumnInfo::new("score".to_string(), DataType::Real, false),
        ]);

        let mut input_scan = TableScanOperator::new("students".to_string(), None, &context).unwrap();

        // 设置测试数据：学生成绩（无序）
        input_scan.set_test_data(vec![
            ExecutionTuple::new(vec![Value::Integer(3), Value::Text("Charlie".to_string()), Value::Real(85.5)]),
            ExecutionTuple::new(vec![Value::Integer(1), Value::Text("Alice".to_string()), Value::Real(92.0)]),
            ExecutionTuple::new(vec![Value::Integer(4), Value::Text("David".to_string()), Value::Real(78.0)]),
            ExecutionTuple::new(vec![Value::Integer(2), Value::Text("Bob".to_string()), Value::Real(88.5)]),
            ExecutionTuple::new(vec![Value::Integer(5), Value::Text("Eve".to_string()), Value::Real(95.0)]),
        ]);

        // 创建排序键：按score降序排序
        let sort_keys = vec![
            SortKey {
                column_index: 2, // score列
                direction: SortDirection::Descending,
            }
        ];

        // 创建排序算子
        let mut sort_op = SortOperator::new(
            Box::new(input_scan),
            sort_keys,
            input_schema,
        );

        // 执行排序
        sort_op.open(&mut context).unwrap();

        let mut results = Vec::new();
        while let Some(tuple) = sort_op.next(&mut context).unwrap() {
            results.push(tuple);
        }

        sort_op.close(&mut context).unwrap();

        // 验证结果：应该按score降序排列
        assert_eq!(results.len(), 5);

        // 验证排序顺序
        let scores: Vec<f64> = results.iter()
            .map(|tuple| match &tuple.values[2] {
                Value::Real(score) => *score,
                _ => panic!("期望Real类型的score"),
            })
            .collect();

        assert_eq!(scores, vec![95.0, 92.0, 88.5, 85.5, 78.0]);

        // 验证对应的学生姓名
        let names: Vec<String> = results.iter()
            .map(|tuple| match &tuple.values[1] {
                Value::Text(name) => name.clone(),
                _ => panic!("期望Text类型的name"),
            })
            .collect();

        assert_eq!(names, vec!["Eve", "Alice", "Bob", "Charlie", "David"]);

        // 验证统计信息
        let stats = sort_op.statistics();
        assert_eq!(stats.tuples_output, 5);
    }

    #[test]
    fn test_sort_operator_multiple_keys() {
        let mut context = create_test_context();

        // 创建输入表扫描
        let input_schema = ExecutionSchema::new(vec![
            ColumnInfo::new("department".to_string(), DataType::Text(Some(50)), false),
            ColumnInfo::new("name".to_string(), DataType::Text(Some(50)), false),
            ColumnInfo::new("salary".to_string(), DataType::Integer, false),
        ]);

        let mut input_scan = TableScanOperator::new("employees".to_string(), None, &context).unwrap();

        // 设置测试数据：员工信息
        input_scan.set_test_data(vec![
            ExecutionTuple::new(vec![Value::Text("IT".to_string()), Value::Text("Bob".to_string()), Value::Integer(70000)]),
            ExecutionTuple::new(vec![Value::Text("HR".to_string()), Value::Text("Alice".to_string()), Value::Integer(60000)]),
            ExecutionTuple::new(vec![Value::Text("IT".to_string()), Value::Text("Charlie".to_string()), Value::Integer(80000)]),
            ExecutionTuple::new(vec![Value::Text("HR".to_string()), Value::Text("David".to_string()), Value::Integer(65000)]),
            ExecutionTuple::new(vec![Value::Text("IT".to_string()), Value::Text("Alice".to_string()), Value::Integer(75000)]),
        ]);

        // 创建排序键：先按department升序，再按salary降序
        let sort_keys = vec![
            SortKey {
                column_index: 0, // department列
                direction: SortDirection::Ascending,
            },
            SortKey {
                column_index: 2, // salary列
                direction: SortDirection::Descending,
            }
        ];

        // 创建排序算子
        let mut sort_op = SortOperator::new(
            Box::new(input_scan),
            sort_keys,
            input_schema,
        );

        // 执行排序
        sort_op.open(&mut context).unwrap();

        let mut results = Vec::new();
        while let Some(tuple) = sort_op.next(&mut context).unwrap() {
            results.push(tuple);
        }

        sort_op.close(&mut context).unwrap();

        // 验证结果
        assert_eq!(results.len(), 5);

        // 验证排序顺序：HR部门在前，IT部门在后，同部门内按薪资降序
        let expected_order = vec![
            ("HR", "David", 65000),   // HR部门，薪资高的在前
            ("HR", "Alice", 60000),   // HR部门，薪资低的在后
            ("IT", "Charlie", 80000), // IT部门，薪资最高
            ("IT", "Alice", 75000),   // IT部门，薪资中等
            ("IT", "Bob", 70000),     // IT部门，薪资最低
        ];

        for (i, (exp_dept, exp_name, exp_salary)) in expected_order.iter().enumerate() {
            let tuple = &results[i];
            assert_eq!(tuple.values[0], Value::Text(exp_dept.to_string()));
            assert_eq!(tuple.values[1], Value::Text(exp_name.to_string()));
            assert_eq!(tuple.values[2], Value::Integer(*exp_salary));
        }

        // 验证统计信息
        let stats = sort_op.statistics();
        assert_eq!(stats.tuples_output, 5);
    }

    #[test]
    fn test_limit_operator() {
        let mut context = create_test_context();

        // 创建输入表扫描
        let input_schema = ExecutionSchema::new(vec![
            ColumnInfo::new("id".to_string(), DataType::Integer, false),
            ColumnInfo::new("value".to_string(), DataType::Text(Some(50)), false),
        ]);

        let mut context = create_test_context();
        let mut input_scan = TableScanOperator::new("test_data".to_string(), None, &context).unwrap();

        // 设置测试数据：10行数据
        input_scan.set_test_data(vec![
            ExecutionTuple::new(vec![Value::Integer(1), Value::Text("A".to_string())]),
            ExecutionTuple::new(vec![Value::Integer(2), Value::Text("B".to_string())]),
            ExecutionTuple::new(vec![Value::Integer(3), Value::Text("C".to_string())]),
            ExecutionTuple::new(vec![Value::Integer(4), Value::Text("D".to_string())]),
            ExecutionTuple::new(vec![Value::Integer(5), Value::Text("E".to_string())]),
            ExecutionTuple::new(vec![Value::Integer(6), Value::Text("F".to_string())]),
            ExecutionTuple::new(vec![Value::Integer(7), Value::Text("G".to_string())]),
            ExecutionTuple::new(vec![Value::Integer(8), Value::Text("H".to_string())]),
            ExecutionTuple::new(vec![Value::Integer(9), Value::Text("I".to_string())]),
            ExecutionTuple::new(vec![Value::Integer(10), Value::Text("J".to_string())]),
        ]);

        // 创建限制算子：LIMIT 3
        let mut limit_op = LimitOperator::with_limit_only(
            Box::new(input_scan),
            3,
            input_schema,
        );

        // 执行限制
        limit_op.open(&mut context).unwrap();

        let mut results = Vec::new();
        while let Some(tuple) = limit_op.next(&mut context).unwrap() {
            results.push(tuple);
        }

        limit_op.close(&mut context).unwrap();

        // 验证结果：应该只有3行
        assert_eq!(results.len(), 3);

        // 验证内容
        for i in 0..3 {
            assert_eq!(results[i].values[0], Value::Integer((i + 1) as i64));
            assert_eq!(results[i].values[1], Value::Text(((b'A' + i as u8) as char).to_string()));
        }

        // 验证统计信息
        let stats = limit_op.statistics();
        assert_eq!(stats.tuples_output, 3);
    }

    #[test]
    fn test_limit_operator_with_offset() {
        let mut context = create_test_context();

        // 创建输入表扫描
        let input_schema = ExecutionSchema::new(vec![
            ColumnInfo::new("id".to_string(), DataType::Integer, false),
            ColumnInfo::new("value".to_string(), DataType::Text(Some(50)), false),
        ]);

        let mut context = create_test_context();
        let mut input_scan = TableScanOperator::new("test_data".to_string(), None, &context).unwrap();

        // 设置测试数据：10行数据
        input_scan.set_test_data(vec![
            ExecutionTuple::new(vec![Value::Integer(1), Value::Text("A".to_string())]),
            ExecutionTuple::new(vec![Value::Integer(2), Value::Text("B".to_string())]),
            ExecutionTuple::new(vec![Value::Integer(3), Value::Text("C".to_string())]),
            ExecutionTuple::new(vec![Value::Integer(4), Value::Text("D".to_string())]),
            ExecutionTuple::new(vec![Value::Integer(5), Value::Text("E".to_string())]),
            ExecutionTuple::new(vec![Value::Integer(6), Value::Text("F".to_string())]),
            ExecutionTuple::new(vec![Value::Integer(7), Value::Text("G".to_string())]),
            ExecutionTuple::new(vec![Value::Integer(8), Value::Text("H".to_string())]),
            ExecutionTuple::new(vec![Value::Integer(9), Value::Text("I".to_string())]),
            ExecutionTuple::new(vec![Value::Integer(10), Value::Text("J".to_string())]),
        ]);

        // 创建限制算子：OFFSET 3 LIMIT 4
        let mut limit_op = LimitOperator::new(
            Box::new(input_scan),
            4, // limit
            3, // offset
            input_schema,
        );

        // 执行限制
        limit_op.open(&mut context).unwrap();

        let mut results = Vec::new();
        while let Some(tuple) = limit_op.next(&mut context).unwrap() {
            results.push(tuple);
        }

        limit_op.close(&mut context).unwrap();

        // 验证结果：应该跳过前3行，然后取4行
        assert_eq!(results.len(), 4);

        // 验证内容：应该是第4-7行（D, E, F, G）
        let expected_values = vec!["D", "E", "F", "G"];
        for i in 0..4 {
            assert_eq!(results[i].values[0], Value::Integer((i + 4) as i64));
            assert_eq!(results[i].values[1], Value::Text(expected_values[i].to_string()));
        }

        // 验证统计信息
        let stats = limit_op.statistics();
        assert_eq!(stats.tuples_output, 4);
    }

    #[test]
    fn test_sort_with_limit() {
        let mut context = create_test_context();

        // 创建输入表扫描
        let input_schema = ExecutionSchema::new(vec![
            ColumnInfo::new("id".to_string(), DataType::Integer, false),
            ColumnInfo::new("score".to_string(), DataType::Real, false),
        ]);

        let mut context = create_test_context();
        let mut input_scan = TableScanOperator::new("scores".to_string(), None, &context).unwrap();

        // 设置测试数据：分数数据
        input_scan.set_test_data(vec![
            ExecutionTuple::new(vec![Value::Integer(1), Value::Real(85.0)]),
            ExecutionTuple::new(vec![Value::Integer(2), Value::Real(92.0)]),
            ExecutionTuple::new(vec![Value::Integer(3), Value::Real(78.0)]),
            ExecutionTuple::new(vec![Value::Integer(4), Value::Real(95.0)]),
            ExecutionTuple::new(vec![Value::Integer(5), Value::Real(88.0)]),
            ExecutionTuple::new(vec![Value::Integer(6), Value::Real(91.0)]),
        ]);

        // 创建排序算子：按score降序
        let sort_keys = vec![
            SortKey {
                column_index: 1, // score列
                direction: SortDirection::Descending,
            }
        ];

        let sort_op = SortOperator::new(
            Box::new(input_scan),
            sort_keys,
            input_schema.clone(),
        );

        // 创建限制算子：取前3名
        let mut limit_op = LimitOperator::with_limit_only(
            Box::new(sort_op),
            3,
            input_schema,
        );

        // 执行组合算子
        limit_op.open(&mut context).unwrap();

        let mut results = Vec::new();
        while let Some(tuple) = limit_op.next(&mut context).unwrap() {
            results.push(tuple);
        }

        limit_op.close(&mut context).unwrap();

        // 验证结果：应该是前3名的分数
        assert_eq!(results.len(), 3);

        // 验证分数顺序：95.0, 92.0, 91.0
        let scores: Vec<f64> = results.iter()
            .map(|tuple| match &tuple.values[1] {
                Value::Real(score) => *score,
                _ => panic!("期望Real类型的score"),
            })
            .collect();

        assert_eq!(scores, vec![95.0, 92.0, 91.0]);

        // 验证对应的ID
        let ids: Vec<i64> = results.iter()
            .map(|tuple| match &tuple.values[0] {
                Value::Integer(id) => *id,
                _ => panic!("期望Integer类型的id"),
            })
            .collect();

        assert_eq!(ids, vec![4, 2, 6]);

        // 验证统计信息
        let stats = limit_op.statistics();
        assert_eq!(stats.tuples_output, 3);
    }
}
